package cn.tedu.csmall.test.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;

public class MyConsumer {
    @Test
    public void consume01() throws Exception {
    //1.创建一个consumer对象 连接nameserver.
        //创建一个消费对象,提供nameserver连接地址 和消费者分组
        DefaultMQPushConsumer consumer=
                new DefaultMQPushConsumer("first-consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        //定义监听队列所属的主题,和过滤的Tags,匹配上的才消费
        consumer.subscribe("first-topic-a","TagA");
    //2.监听队列,实现消费.队列是从主题中获取的.
        consumer.setMessageListener(new MessageListenerConcurrently() {
            /**
             * 每次拿到消息,都调用consumeMessage方法
             * @param list 消息列表,包含消费者拿到的消息对象,每次列表里,只包装一个消息
             * @param consumeConcurrentlyContext 消费环境,消费逻辑的上下文对象.返回消费结果
             * @return 返回值就是告知rocketmq当前消息消费是成功还是失败
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取消息解析
                System.out.println("当前消费的消息个数:"+list.size());
                MessageExt messageExt = list.get(0);
                //拿到消息数据 主要关心的body
                byte[] body = messageExt.getBody();
                String msgStr=new String(body, StandardCharsets.UTF_8);
                messageExt.getMsgId();
                //执行消费逻辑
                System.out.println("消费者拿到消费信息:"+msgStr);
                //返回成功失败
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //建立和nameserver长链接
        consumer.start();
        while(true);
    }

    /**
     * 这是第二组消费者,和第一组消费者绑定消费同一个主题
     * 所以,消息偏移量在主题中重新计算
     * @throws Exception
     */
    @Test
    public void consume02() throws Exception {
        DefaultMQPushConsumer consumer=
                new DefaultMQPushConsumer("second-consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("first-topic-a","TagA");
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("当前消费的消息个数:"+list.size());
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String msgStr=new String(body, StandardCharsets.UTF_8);
                messageExt.getMsgId();
                System.out.println("消费者拿到消费信息:"+msgStr);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        while(true);
    }

    /**
     * 测试延迟消息消费,是否在延迟3级中 10s之后处理
     * @throws Exception
     */
    @Test
    public void consume03() throws Exception {
        //修改分组
        DefaultMQPushConsumer consumer=
                new DefaultMQPushConsumer("third-consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        //修改主题,和匹配tags *
        consumer.subscribe("second-topic-b","*");
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("当前消费的消息个数:"+list.size());
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String msgStr=new String(body, StandardCharsets.UTF_8);
                messageExt.getMsgId();
                //打桩,添加时间,检查延迟时间段
                System.out.println("消费者拿到消费信息:"+msgStr+ LocalDateTime.now());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        while(true);
    }

}
